<?php

declare(strict_types=1);

namespace JfbQueue\Driver;

use Hyperf\AsyncQueue\Driver\ChannelConfig;
use Hyperf\Context\ApplicationContext;
use Hyperf\Contract\ConfigInterface;
use Hyperf\Process\Annotation\Process;
use Hyperf\Process\ProcessManager;
use Hyperf\Redis\RedisFactory;
use Hyperf\Redis\RedisProxy;
use Hyperfx\Framework\Exception\AppException;
use Hyperfx\Framework\Logger\Logx;
use JfbQueue\JobXMessage;
use JfbQueue\Message\RedisStreamMessage;
use Hyperfx\Utils\AppEnvUtil;
use Hyperfx\Utils\RunModeUtil;
use Lysice\HyperfRedisLock\RedisLock;
use Psr\Container\ContainerInterface;
use Hyperf\Process\AbstractProcess;

class RedisStreamDriver implements DriverInterface {


    protected RedisProxy $redis;
    private array $stream = [];
    private string $group = '';
    private string $consumer;
    private int $block;
    private int $maxAttempts = 5;
    private int $handleTimeout = 10 * 1000;
    private string $failStream = '';
    public function __construct(ContainerInterface $container, protected array $config)
    {
        $this->redis = $container->get(RedisFactory::class)->get($config['redis']['pool'] ?? 'default');
        $this->stream = $config['stream'];
        $this->group = $this->config['group'];
        $this->maxAttempts = $this->getMaxAttempts();

        $handleTimeout = $this->config['handle_timeout'] ?? 10;
        if ($handleTimeout > 0) {
            $this->handleTimeout = $handleTimeout * 1000;
        }
        $intWaitSeconds = $this->config['wait_seconds'] ?? 5;
        if ($intWaitSeconds < 1) {
            $intWaitSeconds = 5;
        }
        $this->block = $intWaitSeconds * 1000;
        $this->failStream = sprintf('%s:fail', $this->stream['key']);
    }

    public function init() {
        $this->consumer = cli_get_process_title();
        // 创建消费者组
        $this->redis->xGroup('CREATE', $this->stream['key'], $this->group, '0');
    }

    public function getMaxAttempts(): int {
        $maxAttempts = $this->config['max_attempts'] ?? 10;
        if ($maxAttempts > 0) {
            return $maxAttempts;
        }
        return 5;
    }
    
    public function push(array $messages, int $delay = 0): array|bool {

        // 获取消息数量
        $count = count($messages);
        if ($count === 0) {
            return [];
        }
        
        // handle function
        /* @var $redis RedisProxy */
        $function = function($redis, $messages) {
            $result = [];
            /* @var $objMessage RedisStreamMessage */
            foreach ($messages as $objMessage) {
                $result[] = $redis->xAdd($this->stream['key'], '*', $objMessage->serialize());
            }
            return $result;
        };

        // 一个消息时
        if ($count == 1) {
            return $function($this->redis, $messages);
        }

        $redis = $this->redis->pipeline();
        $function($redis, $messages);
        $results = $redis->exec();
        if (empty($results)) {
            return false;
        }
        foreach ($results as $result) {
            if (empty($result)) {
                return false;
            }
        }
        return $results;
    }

    public function pop(): array|bool {
        $data = $this->redis->xReadGroup($this->group, $this->consumer, [$this->stream['key'] => $this->stream['id']], $this->config['batch_size'] ?? 1, $this->block);
        if (false === $data) {
            // 报错时，避免循环太快
            sleep(5);
            return false;
        }
        $messages = $data[$this->stream['key']] ?? [];
        if (empty($messages)) {
            return false;
        }
        $result = [];
        foreach ($messages as $id => $message) {
            $result[] = RedisStreamMessage::unserialize($id, $message);
        }
        return $result;
    }

    public function ack(array $messages): bool {
        $arrMsgIds = [];
        /* @var $objMessage RedisStreamMessage */
        foreach ($messages as $stream => $objMessage) {
            $arrMsgIds[] = $objMessage->getId();
        }
        $this->redis->multi()
            ->xAck($this->stream['key'], $this->group, $arrMsgIds)
            ->xDel($this->stream['key'], $arrMsgIds)
            ->exec();
        return true;
    }


    private function getPendingInfo() {
        $pending = $this->redis->xPending($this->stream['key'], $this->group);
        if (empty($pending)) {
            return [0, '-', '+'];
        }
        $total = (int) $pending[0] ?? 0;
        $startId = $pending[1] ?? '-';
        $endId = $pending[2] ?? '+';
        return [$total, $startId, $endId];
    }
    /**
     * 把消息重新分配到消息队列里 3秒内只允许重载一次
     */
    public function reload(): bool {
        $redisLock = new RedisLock($this->redis, sprintf('%s:reload-lock', $this->stream['key']), 3);
        if ($redisLock->acquire()) {
            return $this->_reload();
        }
        return false;
    }
    public function _reload(): bool {
        $count = 10;

        // 获取消息总数
        [$total, $startId, $endId] = $this->getPendingInfo();
        if ($total === 0) {
            return true;
        }

        while (true) {
            $items = $this->redis->rawCommand('xpending', $this->stream['key'], $this->group, 'IDLE', $this->handleTimeout, $startId, '+', $count);
            if (empty($items)) {
                break;
            }

            // 是否还有下一页
            $isHasNextPage = count($items) === $count;

            // 重新添加的消息ID
            $arrReAddMsgIds = [];
            // 删除的消息ID
            $arrDelMsgIds = [];

            foreach ($items as $item) {
                $startId = $msgId = $item[0];
                $ttl = $item[2];
                // 如果未超时则跳过
                if ($this->handleTimeout > $ttl) {
                    continue;
                }
                // 获取消息
                $ret = $this->redis->xRange($this->stream['key'], $msgId, $msgId, 1);
                $message = $ret[$msgId] ?? [];
                if (empty($message)) {
                    continue;
                }

                // 超过最大限制时则标记为ack并删除
                $attempts = ((int) $message['attempts']);
                if ($attempts > ($this->maxAttempts * 2)) {
                    // 删除任务任务
                    $this->redis->multi()
                        ->xAdd($this->failStream, '*', $message)
                        ->xAck($this->stream['key'], $this->group, [$msgId])
                        ->xDel($this->stream['key'], [$msgId])
                        ->exec();

                    $arrDelMsgIds[] = $msgId;
                    continue;
                }

                $message['attempts'] = $attempts + 1;

                // 重新添加任务至队列
                $this->redis->multi()
                    ->xAdd($this->stream['key'], '*', $message)
                    ->xAck($this->stream['key'], $this->group, [$msgId])
                    ->xDel($this->stream['key'], [$msgId])
                    ->exec();

                $arrReAddMsgIds[] = $msgId;
            }

            if (!empty($arrReAddMsgIds) || !empty($arrDelMsgIds)) {
                Logx::get()->warning('reload message', [
                    'key' => $this->stream['key'],
                    'group' => $this->group,
                    'readd_ids' => $arrReAddMsgIds,
                    'delete_ids' => $arrDelMsgIds,
                ]);
            }

            if (!$isHasNextPage) {
                break;
            }
        }
        return true;
    }

    /**
     * Push a job to failed queue.
     */
    public function fail(array $messages): bool {
        return true;
    }

    public function retry(array $messages, int $retryTime = 5): bool {
        return true;
    }

    /**
     * query queue length
     */
    public function len(): int {
        return (int) $this->redis->xLen($this->stream['key']);
    }
}